package com.heima.article.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.KafkaStreamListener;
import com.heima.model.common.constants.HotArticleConstants;
import com.heima.model.mess.app.ArticleVisitStreamMess;
import com.heima.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {

        return stream.flatMapValues((x)->{
            log.info("进入这个解析当前UpdateArticleMess的方法");
            UpdateArticleMess articleMess = JSON.parseObject(x, UpdateArticleMess.class);
            return Arrays.asList(articleMess.getArticleId() + ":" + articleMess.getType());
        })
                .groupBy((key,value)->value)
                .windowedBy(TimeWindows.of(5000))
                .count().toStream().map((key, value) -> {
                    log.info("***********************************************************************************");
                    log.info("最终结果:key:{}-------------value:{}",key.key().toString(),value.toString());
                    return new KeyValue<String,String>(key.key().toString(), formatObj(key.key().toString(),value.toString()));
                });


    }

    private String formatObj(String key, String value) {
        ArticleVisitStreamMess visitStreamMess = new ArticleVisitStreamMess();
        String[] split = key.split(":");
        visitStreamMess.setArticleId(Long.valueOf(split[0]));

        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
            visitStreamMess.setView(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
            visitStreamMess.setLike(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])) {
            visitStreamMess.setComment(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
            visitStreamMess.setCollect(Long.valueOf(value));
        }

        log.info("formatObj aggs score ArticleVisitStreamMess:{}", visitStreamMess);
        String valueToListener = JSON.toJSONString(visitStreamMess);
        log.info("&&&&&&&&&&&&&&&&&&&&&&&&&&打印发送给kafkaStream的监听结果&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&");
        log.info(valueToListener);
        log.info("&&&&&&&&&&&&&&&&&&&&&&&&&&打印发送结果结束&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&");
        return valueToListener;

    }
}